<?php
namespace app\mqtt\controller;
use Bluerhinos\phpMQTT;
use think\Cache;
use think\Request;

/**
 * 功能说明: MQTT消息处理类，功能为监听all/# 的所有设备消息，根据消息类型进行处理和转发
 * Company:物联网应用专家---成都亿佰特电子科技有限公司
 * User: 郑文军
 * Time: 2021-10-13 16:15
 * version：1.0
 */
class Mqtt extends Base
{
    public function __construct(Request $request = null)
    {

        parent::__construct($request);
    }

    public function index()
    {
        ob_start(); //打开缓冲区getStatusRes
        ob_end_flush();
        set_time_limit(0);
        $topics['$SYS/brokers/+/clients/#'] = array("qos" => 0, "function" => array($this, "onMessage"));
        $this->mqtt->subscribe($topics, 0);
        /* $topics['$SYS/brokers/${node}/clients/${clientId}/disconnected'] = array("qos" => 0, "function" => array($this, "onMessage"));
         $this->mqtt->subscribe($topics, 0);*/
        $topics['all/#'] = array("qos" => 0, "function" => array($this, "onMessage"));
        $this->mqtt->subscribe($topics, 0);
        echo "订阅成功";
        while ($this->mqtt->proc()) { //无限去请求

        }
        $this->mqtt->close();

    }

    function onMessage($topic, $msg)
    {//输出消息
        $message = json_decode($msg, true);
        if (isset($message['disconnected_at']) || isset($message['connected_at'])) {
            $this->OpenOrClose($topic, $message);
        } else {
            //echo "接收到数据消息".$msg;
            $this->sendDate($topic, $message);
        }
        echo "\r\n";

    }

    public function SendMessage($topic, $message)
    {
        $this->mqtt->publish($topic, $message, 0);
    }

    /**
     * 功能说明 设备连接或断开连接处理
     * @param $topic
     * @param $message
     * author 郑
     * time 2021-10-13 16:14
     */
    public function OpenOrClose($topic, $message)
    {
        //接收到上线消息
        $type = 0;
        if (isset($message['connected_at'])) {

            echo "message";
            $res = db('device')->where(['code' => $message["clientid"], 'delete' => 0])->find();
            echo date("Y-m-d H:i:s") . ":接收到消息: 归属用户" . $res["uid"] . " 设备ID:" . $message["clientid"] . "已连接";
            db('device')->where(['id' => $res["id"]])->update(['start' => 1]);

            //更新最近上线时间
            db("device")->where(['id' => $res['id']])->update(['last_login_time' => time()]);

            //给前端推送上线
            $newMessage = ['id' => $message['clientid'], 'name' => $res['name'], 'type' => 'connect', 'uid' => $res['uid']];
            $this->SendMessage("user/" . $res['uid'], json_encode($newMessage));
            $type = 1;

        } //接受到离线消息
        else if (isset($message['disconnected_at'])) {
            echo "接收到消息: 归属用户" . $message["clientid"] . " 设备ID:" . $message["clientid"] . "已断开连接";
            $res = db('device')->where(['code' => $message["clientid"], 'delete' => 0])->find();
            db('device')->where(['id' => $res["id"]])->update(['start' => 0]);

            //给前端推送下线
            $newMessage = ['id' => $message['clientid'], 'name' => $res['name'], 'type' => 'disconnect', 'uid' => $res['uid']];
            $this->SendMessage("user/" . $res['uid'], json_encode($newMessage));
            $type = 2;
        }


        //记录上下线日志
        $date = [];
        $date['create_time'] = time();
        $date['params'] = json_encode($message);
        $date['uid'] = $res['uid'];
        $date['code'] = $res['code'];
        $date['did'] = $res['id'];
        $date['type'] = $type;
        db("device_log")->insert($date);

    }


    /**
     * 功能说明 数据消息处理
     * @param $topic
     * @param $message
     * author 郑
     * time 2021/3/17 9:20
     */
    public function sendDate($topic, $message)
    {
        //接受到数据传输
        if (!isset($message['method'])) {
            return;
        }


        //系统信息上报
        if ($message['method'] == 'postSysInfo' || $message['method'] == 'reqSysInfoRsp') {
            $this->postSysInfo($message);
        }

        //参数配置回复信息
        if ($message['method'] == 'cfgSysParamsRsp' || $message['method'] == "getSysParamsRsp") {
            $this->cfgSysParamsRsp($message);
        }

        //心跳数据上报
        if ($message['method'] == "keepAlive") {
            $this->keepAlive($message);
        }

        //控制配置指令回复
        if ($message['method'] == 'setDOStatusRsp' || $message['method'] == "getDOStatusRsp"
            || $message['method'] == 'postDIInfo' || $message['method'] == "getDIStatusRsp" || $message['method'] == 'cfgDIParamsRsp'
            || $message['method'] == 'getAIStatusRsp' || $message['method'] == "reqAIParamsRsp" || $message['method'] == "cfgAIParamsRsp"
            || $message['method'] == 'setPWMStatusRsp' || $message['method'] == "cfgPWMParamsRsp" || $message['method'] == "reqPWMParamsRsp"
            || $message['method'] == "cfgDOParamsRsp" || $message['method'] == "reqDOParamsRsp"
            || $message['method'] == "cfgUARTParamsRsp" || $message['method'] == "reqUARTParamsRsp"
            || $message['method'] == "revUARTData"
        ) {
            $this->settingRsp($message);
        }
    }

    public function settingRsp($message)
    {
        $params = $message['params'];
        $device = db('device')->where(['code' => $message['sn']])->find();
        if (!$device) {
            return;
        }
        //更新DO数据库参数
        if ($message['method'] == "cfgDOParamsRsp" || $message['method'] == "reqDOParamsRsp") {
            foreach ($params as $v) {
                $date = ['mode' => $v['mode'], 'isPost' => $v['isPost'], 'pulseTime' => $v['pulseTime']];
                db("device_do")->where(['did' => $device['id'], 'num' => $v['num']])->update($date);
            }
        } else if ($message['method'] == "getDOStatusRsp") {
            foreach ($params as $v) {
                $date = ['value' => $v['value']];
                db("device_do")->where(['did' => $device['id'], 'num' => $v['num']])->update($date);
            }
        } else if ($message['method'] == "getDIStatusRsp" || $message['method'] == "postDIInfo") {
            foreach ($params as $v) {
                $date = ['value' => $v['value']];
                db("device_di")->where(['did' => $device['id'], 'num' => $v['num']])->update($date);
            }
        } //更新DI数据库状态
        else if ($message['method'] == "cfgDIParamsRsp") {
            foreach ($params as $v) {
                $date = ['isPost' => $v['isPost']];
                db("device_di")->where(['did' => $device['id'], 'num' => $v['num']])->update($date);
            }
        } //更新AI数据库配置参数
        else if ($message['method'] == "cfgAIParamsRsp" || $message['method'] == "reqAIParamsRsp") {
            foreach ($params as $v) {
                $date = ['isPost' => $v['isPost']];
                db("device_ai")->where(['did' => $device['id'], 'num' => $v['num']])->update($date);
            }
        } //更新PWM配置信息
        else if ($message['method'] == "cfgPWMParamsRsp" || $message['method'] == "reqPWMParamsRsp") {
            foreach ($params as $v) {
                $date = ['frqDiv' => $v['frqDiv'], 'highLevelCnt' => $v['highLevelCnt'], 'totalCnt' => $v['totalCnt']];
                db("device_pwm")->where(['did' => $device['id'], 'num' => $v['num']])->update($date);
            }
        } //更新PWM状态信息
        else if ($message['method'] == "setPWMStatusRsp") {
            foreach ($params as $v) {
                $date = ['value' => $v['status']];
                db("device_pwm")->where(['did' => $device['id'], 'num' => $v['num']])->update($date);
            }
        } //更新UART配置信息
        else if ($message['method'] == "cfgUARTParamsRsp" || $message['method'] == "reqUARTParamsRsp") {
            foreach ($params as $v) {
                $date = ['baud' => $v['baud'], 'parity' => $v['parity'], 'stop' => $v['stop'], "dataBits" => $v['dataBits']];
                db("device_uart")->where(['did' => $device['id'], 'num' => $v['num']])->update($date);
            }
        }


        //给前端推送消息
        $device = db('device')->where(['code' => $message["sn"], 'delete' => 0])->find();
        $newMessage = ['method' => $message['method'], 'code' => $device['code'], 'params' => $message['params']];
        echo "发送:" . "user/" . $device['uid'] . '/' . $device['code'];
        echo "推送消息:" . json_encode($newMessage);
        $this->SendMessage("user/" . $device['uid'] . '/' . $device['code'], json_encode($newMessage));
    }

    /**
     * 功能说明 系统参数配置
     * @param $message
     * author 郑
     * time 2021-10-11 9:55
     */
    public function cfgSysParamsRsp($message)
    {
        $res = db('device')->where(['code' => $message["sn"], 'delete' => 0])->find();
        if (!$res) {
            return;
        }
        $param = $message['params'];
        //更新心跳周期和自定义消息内容
        $date['keepAliveTime'] = $param['keepAliveTime'];
        $date['keepAliveDate'] = $param['data'];
        db("device")->where(['id' => $res['id']])->update($date);
    }

    /**
     * 功能说明 心跳数据上报
     * @param $message
     * author 郑
     */
    public function keepAlive($message)
    {
        //直接进行消息存储和转发
        $res = db('device')->where(['code' => $message["sn"], 'delete' => 0])->find();
        if (!$res) {
            return;
        }
        //更新设备当前信息
        $param = $message['params'];
        $data = [];
        $data['version'] = $param['version'];
        $data['iccid'] = $param['ICCID'];
        $data['csq'] = $param['CSQ'];
        $data['imei'] = $param['IMEI'];
        $data['latitude'] = $param['latitude'];
        $data['longitude'] = $param['longitude'];
        db("device")->where(['id' => $res['id']])->update($data);

        //记录设备经纬度信息
        $date = [];
        $date['create_time'] = time();
        $date['uid'] = $res['uid'];
        $date['code'] = $res['code'];
        $date['did'] = $res['id'];
        $date['long'] = $param['longitude'];
        $date['lat'] = $param['latitude'];
        db("device_address")->insert($date);


        //更新DI
        if (!empty($param['DIStatus'])) {
            foreach ($param['DIStatus'] as $v) {
                $where = ['num' => $v['num'], 'did' => $res['id']];
                $data = [];
                $data['num'] = $v['num'];
                $data['value'] = $v['value'];
                $data['did'] = $res['id'];
                $isExist = db("device_di")->where($where)->find();
                if ($isExist) {
                    db("device_di")->where($where)->update($data);
                } else {
                    $data['name'] = "DI" . $v['num'];
                    db("device_di")->insert($data);
                }
            }
        }


        //更新DO
        if (!empty($param['DOStatus'])) {
            foreach ($param['DOStatus'] as $v) {
                $where = ['num' => $v['num'], 'did' => $res['id']];
                $data = [];
                $data['num'] = $v['num'];
                $data['value'] = $v['value'];
                $data['did'] = $res['id'];
                $isExist = db("device_do")->where($where)->find();
                if ($isExist) {
                    $data['name'] = "DO" . $v['num'];
                    db("device_do")->where($where)->update($data);
                } else {
                    $data['name'] = "DO" . $v['num'];
                    db("device_do")->insert($data);
                }
            }
        }


        //更新AI
        if (!empty($param['AIStatus'])) {
            foreach ($param['AIStatus'] as $v) {
                $where = ['num' => $v['num'], 'did' => $res['id']];
                $data = [];
                $data['num'] = $v['num'];
                $data['value'] = $v['value'];
                $data['did'] = $res['id'];
                $isExist = db("device_ai")->where($where)->find();
                if ($isExist) {
                    $data['name'] = "AI" . $v['num'];
                    db("device_ai")->where($where)->update($data);
                } else {
                    $data['name'] = "AI" . $v['num'];
                    db("device_ai")->insert($data);
                }
            }
        }

        //更新PWM
        if (!empty($param['PWMStatus'])) {
            foreach ($param['PWMStatus'] as $v) {
                $where = ['num' => $v['num'], 'did' => $res['id']];
                $data = [];
                $data['num'] = $v['num'];
                $data['value'] = $v['value'];
                $data['did'] = $res['id'];
                $isExist = db("device_pwm")->where($where)->find();
                if ($isExist) {
                    $data['name'] = "PWM" . $v['num'];
                    db("device_pwm")->where($where)->update($data);
                } else {
                    $data['name'] = "PWM" . $v['num'];
                    db("device_pwm")->insert($data);
                }
            }
        }

        //更新Uart
        if (!empty($param['UartStatus'])) {
            foreach ($param['UartStatus'] as $v) {
                $where = ['num' => $v['num'], 'did' => $res['id']];
                $data = [];
                $data['num'] = $v['num'];
                $data['value'] = $v['value'];
                $data['did'] = $res['id'];
                $isExist = db("device_uart")->where($where)->find();
                if ($isExist) {
                    $data['name'] = "UART" . $v['num'];
                    db("device_uart")->where($where)->update($data);
                } else {
                    $data['name'] = "UART" . $v['num'];
                    db("device_uart")->insert($data);
                }
            }
        }
    }


    /**
     * 功能说明 更新设备信息
     * @param $message
     * author 郑
     * time 2021-10-11 9:55
     */
    public function postSysInfo($message)
    {
        //直接进行消息存储和转发
        $res = db('device')->where(['code' => $message["sn"], 'delete' => 0])->find();
        if (!$res) {
            return;
        }
        //更新设备当前信息
        $param = $message['params'];
        $data = [];
        $data['model'] = $param['model'];
        $data['version'] = $param['version'];
        $data['iccid'] = $param['ICCID'];
        $data['csq'] = $param['CSQ'];
        $data['imei'] = $param['IMEI'];
        $data['latitude'] = $param['latitude'];
        $data['longitude'] = $param['longitude'];
        db("device")->where(['id' => $res['id']])->update($data);

        //记录设备经纬度信息
        $date = [];
        $date['create_time'] = time();
        $date['uid'] = $res['uid'];
        $date['code'] = $res['code'];
        $date['did'] = $res['id'];
        $date['long'] = $param['longitude'];
        $date['lat'] = $param['latitude'];
        db("device_address")->insert($date);

        //更新DI
        if (!empty($param['DIChTab'])) {
            foreach ($param['DIChTab'] as $v) {

                $where = ['num' => $v, 'did' => $res['id']];
                $data = [];
                $data['num'] = $v;
                $data['value'] = 1;
                $data['did'] = $res['id'];
                $isExist = db("device_di")->where($where)->find();
                if ($isExist) {
                    db("device_di")->where($where)->update($data);
                } else {
                    $data['name'] = "DI" . $v;
                    db("device_di")->insert($data);
                }
            }
        }


        //更新DO
        if (!empty($param['DOChTab'])) {
            foreach ($param['DOChTab'] as $v) {
                $where = ['num' => $v, 'did' => $res['id']];
                $data = [];
                $data['num'] = $v;
                $data['value'] = 1;
                $data['did'] = $res['id'];
                $isExist = db("device_do")->where($where)->find();
                if (!$isExist) {
                    $data['name'] = "DO" . $v;
                    db("device_do")->insert($data);
                }
            }
        }


        //更新AI
        if (!empty($param['AIChTab'])) {
            foreach ($param['AIChTab'] as $v) {
                $where = ['num' => $v, 'did' => $res['id']];
                $data = [];
                $data['num'] = $v;
                $data['value'] = 1;
                $data['did'] = $res['id'];
                $isExist = db("device_ai")->where($where)->find();
                if (!$isExist) {
                    $data['name'] = "AI" . $v;
                    db("device_ai")->insert($data);
                }
            }
        }

        //更新PWM
        if (!empty($param['PWMChTab'])) {
            foreach ($param['PWMChTab'] as $v) {
                $where = ['num' => $v, 'did' => $res['id']];
                $data = [];
                $data['num'] = $v;
                $data['value'] = 1;
                $data['did'] = $res['id'];
                $isExist = db("device_pwm")->where($where)->find();
                if (!$isExist) {
                    $data['name'] = "PWM" . $v;
                    db("device_pwm")->insert($data);
                }
            }
        }

        //更新Uart
        if (!empty($param['UartChTab'])) {
            foreach ($param['UartChTab'] as $v) {
                $where = ['num' => $v, 'did' => $res['id']];
                $data = [];
                $data['num'] = $v;
                $data['value'] = 1;
                $data['did'] = $res['id'];
                $isExist = db("device_uart")->where($where)->find();
                if (!$isExist) {
                    $data['name'] = "UART" . $v;
                    db("device_uart")->insert($data);
                }
            }
        }
        $topic = "all/" . $res['code'] . '/sub';

        //查询系统参数信息
        $selectMessage = [
            "msgId" => date("YmdHis") . rand(100000, 999999),
            "method" => "getSysParams",
            "sn" => $res['code'],
        ];
        $this->SendMessage($topic, json_encode($selectMessage));


        //查询DO配置
        $selectMessage = [
            "msgId" => date("YmdHis") . rand(100000, 999999),
            "method" => "reqDOParams",
            "sn" => $res['code'],
            "params" => ['chTab' => $param['DOChTab']]
        ];
        $this->SendMessage($topic, json_encode($selectMessage));
        //查询DO状态getDOStatus
        $selectMessage['method'] = "getDOStatus";
        $selectMessage['msgId'] = date("YmdHis") . rand(100000, 999999);
        $this->SendMessage($topic, json_encode($selectMessage));


        //查询DI信息
        $selectMessage = [
            "msgId" => date("YmdHis") . rand(100000, 999999),
            "method" => "reqDIParams",
            "sn" => $res['code'],
            "params" => ['chTab' => $param['DIChTab']]
        ];
        $this->SendMessage($topic, json_encode($selectMessage));

        //查询DI状态 getDIStatus
        $selectMessage['method'] = "getDIStatus";
        $selectMessage['msgId'] = date("YmdHis") . rand(100000, 999999);
        $this->SendMessage($topic, json_encode($selectMessage));

        //查询AI配置信息
        $selectMessage = [
            "msgId" => date("YmdHis") . rand(100000, 999999),
            "method" => "reqAIParams",
            "sn" => $res['code'],
            "params" => ['chTab' => $param['AIChTab']]
        ];
        $this->SendMessage($topic, json_encode($selectMessage));
        $selectMessage['method'] = "getAIStatus";
        $selectMessage['msgId'] = date("YmdHis") . rand(100000, 999999);
        $this->SendMessage($topic, json_encode($selectMessage));

        //查询PWM配置信息
        $selectMessage = [
            "msgId" => date("YmdHis") . rand(100000, 999999),
            "method" => "reqPWMParams",
            "sn" => $res['code'],
            "params" => ['chTab' => $param['PWMChTab']]
        ];
        $this->SendMessage($topic, json_encode($selectMessage));
        $selectMessage['method'] = "getPWMStatus";
        $selectMessage['msgId'] = date("YmdHis") . rand(100000, 999999);
        $this->SendMessage($topic, json_encode($selectMessage));

        //查询UART参数配置信息
        $selectMessage = [
            "msgId" => date("YmdHis") . rand(100000, 999999),
            "method" => "reqUARTParams",
            "sn" => $res['code'],
            "params" => ['chTab' => $param['UartChTab']]
        ];
        $this->SendMessage($topic, json_encode($selectMessage));


    }


    /**
     * 功能说明
     * @param $message
     * author 郑
     * time 2021-06-11 17:37
     */
    public function Response($message)
    {
        $res = db("task_device")->where(['taskId' => $message['taskId'], 'code' => $message['id'], 'status' => 2])->find();
        if ($res) {
            if ($message['code'] == 1) {
                $status = 4;
            } else if ($message['code'] == 2) {
                $status = 3;
            }
            db('task_device')->where(['id' => $res['id']])->update(['status' => $status]);
        }
    }


    /**
     * 给内部服务器推送消息
     * @param $host 服务器地址
     * @param $port 服务器端口
     * @param $str 推送内容
     * @param int $back
     * @return bool|string 是否推送成功
     * time 2021-10-08 16:16
     */
    function sendSocketMsg($host, $port, $str, $back = 0)
    {
        $socket = socket_create(AF_INET, SOCK_STREAM, 0);
        if ($socket < 0) return false;
        $result = @socket_connect($socket, $host, $port);
        if ($result == false) return false;
        socket_write($socket, $str, strlen($str));
        if ($back != 0) {
            $input = socket_read($socket, 1024);
            socket_close($socket);
            return $input;
        } else {
            socket_close($socket);
            return true;
        }
    }
}





